#include <atomic>
#include <chrono>
#include <condition_variable>
#include <functional>
#include <iostream>
#include <memory>
#include <mutex>
#include <queue>
#include <semaphore>
#include <thread>
#include <vector>
using ThreadFunc = std::function<void()>;

class Result;
class Any;
class Task {
 public:
  Task() : result_(nullptr) {}
  ~Task() = default;
  virtual Any Run() = 0;
  void SetResult(Result* result) {
    if (result != nullptr) result_ = result;
  }
  void Exec();

 private:
  // std::shared_ptr<Result> result_;造成循环引用问题
  // TODO:使用weak_ptr
  Result* result_;
  // Result类生命周期比Task长
};

// -----------------------------------
class Any {
 public:
  Any() = default;
  ~Any() = default;
  Any(const Any&) = delete;
  Any& operator=(const Any&) = delete;
  Any(Any&&) = default;
  Any& operator=(Any&&) = default;

  template <typename T>
  Any(T data) : base_(std::make_unique<Derived<T>>(data)) {}

  template <typename T>
  T Cast() {
    Derived<T>* ptr = dynamic_cast<Derived<T>*>(base_.get());
    if (ptr == nullptr) {
      throw " type is unmatched!";
    }
    return ptr->data_;
  }

 private:
  class Base {
   public:
    virtual ~Base() = default;
  };
  template <typename T>
  class Derived : public Base {
   public:
    Derived(T data) : data_(data) {}
    ~Derived() = default;
    T data_;
  };

  std::unique_ptr<Base> base_;
};

// ------------------
// TODO:是使用的二元信号量吗？
class Semaphore {
 public:
  Semaphore(int rlimit = 0) : rlimit_(rlimit) {}
  ~Semaphore() = default;
  void Wait();
  void Post();

 private:
  std::atomic_int rlimit_;
  std::mutex mtx_;
  std::condition_variable cond_;
};
void Semaphore::Wait() {
  std::unique_lock<std::mutex> ul(mtx_);
  cond_.wait(ul, [&]() -> bool { return rlimit_ > 0; });
  rlimit_--;
}
void Semaphore::Post() {
  std::unique_lock<std::mutex> ul(mtx_);
  rlimit_++;
  cond_.notify_one();
  // cond_.notify_all() TODO:这里用这个还是用那个？
}
// -----------------------
class Result {
 public:
  Result(std::shared_ptr<Task> task, bool flag = true)
      : task_(task), is_valid_(flag) {
    task_->SetResult(this);  // *************************
  }
  ~Result() = default;
  // 因为Any类的拷贝构造和拷贝赋值运算符已删除，所以Result类也只能删除
  Result(const Result&) = delete;
  Result& operator=(const Result&) = delete;
  // Result(Result&&) = default;
  // Result& operator=(Result&&) = default;  // 移动赋值运算符
  // Result&& operator=(const Result&&) = default; 错误

  void Set(Any&& any);
  Any Get();

 private:
  Any any_;
  Semaphore sem_;
  std::atomic_bool is_valid_;

  std::shared_ptr<Task> task_;  // 指向对应获取返回值的任务对象
};

void Result::Set(Any&& any) {
  any_ = std::move(any);
  sem_.Post();
}
Any Result::Get() {
  if (!is_valid_) return nullptr;
  sem_.Wait();
  return std::move(any_);
}

class Thread {
 public:
  Thread(ThreadFunc func) : func_(std::move(func)) {}
  ~Thread() {}
  void Start();

 private:
  ThreadFunc func_;
  std::unique_ptr<std::thread> thread_;
};
class ThreadPool {
 public:
  ThreadPool(int task_que_max_num = 256) : task_que_max_num_(task_que_max_num) {
    task_num_ = 0;
  }
  ~ThreadPool() {}
  ThreadPool(const ThreadPool&) = delete;
  ThreadPool operator=(const ThreadPool&) = delete;

  void Start();
  void SetThreadNum(int thread_init_num = -1);
  Result AddTask(std::shared_ptr<Task> sp);

 private:
  void ThreadRunFunc();

 private:
  std::vector<std::unique_ptr<Thread>> thread_vec_;
  int thread_init_num_;
  ThreadFunc func_;

  std::queue<std::shared_ptr<Task>> task_que_;
  std::atomic_int task_num_;
  int task_que_max_num_;

  std::mutex mtx_;
  std::condition_variable producer_cond_;
  std::condition_variable consumer_cond_;
};

void Thread::Start() {
  thread_ = std::make_unique<std::thread>(func_);
  thread_->detach();
}
// 用户向线程池提交任务,生产！
Result ThreadPool::AddTask(std::shared_ptr<Task> sp) {
  std::unique_lock<std::mutex> ul(mtx_);
  if (!producer_cond_.wait_for(ul, std::chrono::seconds(1), [&]() -> bool {
        return task_num_ < task_que_max_num_;
      })) {
    std::cerr << "task_que is full, AddTask failed!" << std::endl;
    return Result(sp, false);
  }
  task_que_.emplace(sp);  // task_que_.emplace(std::move(sp));
  task_num_++;
  std::cout << "AddTask successfully!" << std::endl;
  consumer_cond_.notify_one();  // FIXME:是否这行代码不用锁住？
  // consumer_cond_.notify_all();
  return Result(sp);
}
// 具体的一个任务执行流程，消费！
void ThreadPool::ThreadRunFunc() {
  while (1) {
    std::shared_ptr<Task> task;
    {
      std::unique_lock<std::mutex> ul(mtx_);
      consumer_cond_.wait(ul, [&]() -> bool { return task_num_ > 0; });
      task = task_que_.front();
      task_que_.pop();
      task_num_--;
      if (task_num_ > 0) consumer_cond_.notify_one();
    }

    if (task != nullptr) {
      task->Exec();
    }
  }
}
void ThreadPool::Start() {
  for (int i = 0; i < thread_init_num_; i++) {
    thread_vec_.emplace_back(std::move(
        std::make_unique<Thread>(std::bind(&ThreadPool::ThreadRunFunc, this))));
  }
  for (const auto& t : thread_vec_) t->Start();
}

void ThreadPool::SetThreadNum(int thread_init_num) {
  if (thread_init_num == -1)
    thread_init_num_ =
        std::thread::hardware_concurrency();  // TODO:1807496520 bug?
  else
    thread_init_num_ = thread_init_num;
}

void Task::Exec() { result_->Set(Run()); }